package com.heima.schedule.service.impl;

import com.heima.common.constants.RedisConstants;
import com.heima.common.constants.ScheduleConstants;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import com.heima.utils.common.BeanHelper;
import com.heima.utils.common.JsonUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.input.Tailer;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * 任务处理业务 实现类
 */
@Service
public class TaskServiceImpl implements TaskService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;


    /**
     * 添加任务
     */
    @Override
    @Transactional
    public Long addTask(Task task) {

        //把任务添加到DB
        addTaskToDb(task);

        //把任务添加Redis
        addTaskToCache(task);

        return task.getTaskId();

    }

    /**
     * 消费任务
     */
    @Override
    @Transactional
    public List<Task> pollTask(Integer taskTopic) {
        String key = RedisConstants.TASK_TOPIC_PREFIX + taskTopic;
        //查询redis中符合执行条件的任务
        Set<String> taskSet = stringRedisTemplate.opsForZSet().rangeByScore(key, 0, System.currentTimeMillis());


        //从redis取出任务
        List<Task> taskList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(taskSet)) {
            for (String taskJson : taskSet) {
                Task task = JsonUtils.toBean(taskJson, Task.class);

                //更新DB数据
                updateTaskToDB(task);

                //删除redis数据
                stringRedisTemplate.opsForZSet().remove(key, taskJson);
                taskList.add(task);
            }

        }
        return taskList;
    }

    /**
     * 更新DB数据
     */
    private void updateTaskToDB(Task task) {
        try {
            //删除任务表记录
            taskinfoMapper.deleteById(task.getTaskId());

            //更新任务日志表记录
            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
            taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);//已执行
            taskinfoLogsMapper.updateById(taskinfoLogs);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }

    }


    /**
     * 把任务添加到Redis缓存中
     */
    private void addTaskToCache(Task task) {

        //判断任务是否是当前时间5分钟内的毫秒值
        long millis = DateTime.now().plusMinutes(5).getMillis();
        if (task.getExecuteTime() <= millis) {
            //将任务存入Redis中
            String key = RedisConstants.TASK_TOPIC_PREFIX + task.getTaskTopic();
            stringRedisTemplate.opsForZSet().incrementScore(key, JsonUtils.toString(task), task.getExecuteTime());
        }
    }


    /**
     * 把任务持久化到DB数据库
     */
    private void addTaskToDb(Task task) {
        try {
            //持久化任务对象
            Taskinfo taskinfo = BeanHelper.copyProperties(task, Taskinfo.class);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoMapper.insert(taskinfo);

            //把新产生的任务ID赋值给Task对象
            task.setTaskId(taskinfo.getTaskId());

            //任务日志表
            TaskinfoLogs taskinfoLogs = BeanHelper.copyProperties(taskinfo, TaskinfoLogs.class);
            taskinfoLogs.setVersion(1);       //版本号乐观锁
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);//任务状态
            taskinfoLogsMapper.insert(taskinfoLogs);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }

    }
}
